package com.glodon.paas.job.agent.task;

import static com.glodon.paas.job.JobConstants.*;
import static com.glodon.paas.job.model.TaskStatus.*;
import static com.glodon.paas.job.util.JobUtil.*;

import javax.jms.JMSException;
import javax.jms.Message;
import javax.jms.MessageListener;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.glodon.paas.job.agent.Agent;
import com.glodon.paas.job.agent.mq.JmsCaller;
import com.glodon.paas.job.agent.worker.Caller;
import com.glodon.paas.job.agent.worker.Worker;
import com.glodon.paas.job.agent.worker.WorkerConfig;
import com.glodon.paas.job.agent.worker.WorkerException;
import com.glodon.paas.job.agent.worker.WorkerFactory;
import com.glodon.paas.job.exception.TaskConvertException;
import com.glodon.paas.job.model.Task;
import com.glodon.paas.job.util.JobUtil;

public abstract class AbstractTaskListener implements MessageListener {
    private static final Logger logger = LoggerFactory.getLogger(AbstractTaskListener.class);

    private WorkerFactory workerFactory;
    private Agent agent;
    private WorkerConfig[] workerConfigs;

    public void onMessage(Message msg) {
        try {
            Task task = Task.createTask(msg);
            Caller caller = new JmsCaller(agent, task);
            try {
                task.setCreateTime(getMessageInTime(msg));
                task.setStartTime(getMessageOutTime(msg));
                task.setStatus(running);
                task.addValue(AGENT, JobUtil.getAddress());
                caller.sendProgress(0, "start");
                Worker worker = getWorker(task, caller);
                onTask(task, worker, caller);
                task.setStatus(success);
                caller.sendProgress(100, "success");
                msg.acknowledge();
            } catch (Exception e) {
                logger.error("on message error", e);
                task.setStatus(failure);
                task.addValue(ERROR_MSG, e.getMessage());
                caller.send();
                // throw new JobRuntimeException(e.getMessage(), e);
            }
        } catch (TaskConvertException e) {
            try {
                // drop message
                Task task = new Task();
                task.setTaskId(msg.getStringProperty(PROPERTY_TASK_ID));
                task.setStatus(failure);
                Caller caller = new JmsCaller(agent, task);
                caller.send();
                logger.error(e.getMessage(), e);
            } catch (JMSException jmsE) {
                logger.error(e.getMessage(), e);
            }
        }
    }

    private Worker getWorker(Task task, Caller caller) {
        String type = task.getTaskType();
        String cmd = null;
        String[] envp = null;
        for (WorkerConfig config : workerConfigs) {
            if (type.equals(config.getType())) {
                cmd = config.getCmd();
                envp = config.getEnvp();
            }
        }
        return workerFactory.createWorker(caller, cmd, envp);
    }

    protected abstract void onTask(Task task, Worker worker, Caller caller) throws WorkerException;

    public void setWorkerFactory(WorkerFactory workerFactory) {
        this.workerFactory = workerFactory;
    }

    public void setAgent(Agent agent) {
        this.agent = agent;
    }

    public Agent getAgent() {
        return agent;
    }

    public void setWorkerConfigs(WorkerConfig[] workerConfigs) {
        this.workerConfigs = workerConfigs;
    }

    public WorkerConfig[] getWorkerConfigs() {
        return workerConfigs;
    }

}
